package Kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

/**
 * @author Lu Ruotong
 * @date 2022/11/19 15:34
 */

public class ReadData {
    public static void main(String[] args) {
        /*
        实例：一个consumer   对比producer（序列化-- StringSerializer）
        1.配置项   kafka broker key value consumer-反序列(StringDeserializer) group_id   数据是从哪个地方读取？或者说怎么推拉或者拉取数据
        2.kafka 创建一个consumer
        3.循环 for while topic test  读取topic数据
         */
        Properties pro = new Properties();
        pro.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "niit01:9092");
        pro.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        pro.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");


        //自定义id姓名
        pro.put(ConsumerConfig.GROUP_ID_CONFIG, "niit18");
        pro.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        //消费者组
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(pro);
        //订阅某个topic
        consumer.subscribe(Arrays.asList("stuInfo"));
        //循环
        while (true) {
            //topic:test  读取topic的数据
            ConsumerRecords<String, String> consumerRecord = consumer.poll(5);
            for (ConsumerRecord<String, String> record : consumerRecord) {
                System.out.println("-----------------------");
                //System.out.println("key: " + record.key());
                System.out.println("value:  " + record.value());
            }
        }
    }
}

